/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

 

package com.sui.bigdata.flink.sql.core.parser;

import com.sui.bigdata.flink.sql.core.enums.ETableType;
import com.sui.bigdata.flink.sql.core.table.TableInfo;
import com.sui.bigdata.flink.sql.core.table.TableInfoParser;
import com.sui.bigdata.flink.sql.core.util.DtStringUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.shaded.curator.org.apache.curator.shaded.com.google.common.collect.Lists;
import org.apache.flink.shaded.guava18.com.google.common.base.Strings;

import java.util.*;

/**
 * Reason:
 * Date: 2019/07/10
 * @author yong_chen@sui.com
 */

public class SqlParser {

    private static final char SQL_DELIMITER = ';';
    private static org.apache.log4j.Logger logger = org.apache.log4j.Logger.getLogger(SqlParser.class);


    private static List<IParser> sqlParserList = Lists.newArrayList(CreateFuncParser.newInstance(),
            InsertSqlParser.newInstance(),CreateTableParser.newInstance(), CreateTmpTableParser.newInstance(), SelectSqlParser.newInstance());


    /**
     * flink support sql syntax
     * CREATE TABLE sls_stream() with ();
     * CREATE (TABLE|SCALA) FUNCTION fcnName WITH com.dtstack.com;
     * insert into tb1 select * from tb2;
     * @param sql
     */
    public static SqlTree parseSql(String sql, List<CreateTableParser.SqlParserResult>  preDealTable ) throws Exception {

        if(StringUtils.isBlank(sql)){
            throw new RuntimeException("sql is not null");
        }


        sql = sql.replaceAll("--.*", "")
                .replaceAll("\r\n", " ")
                .replaceAll("\n", " ")
                .replace("\t", " ").trim();

        List<String> sqlArr = DtStringUtil.splitIgnoreQuota(sql, SQL_DELIMITER);
        SqlTree sqlTree = new SqlTree();
        TableInfoParser tableInfoParser = new TableInfoParser();
        for(String childSql : sqlArr){
            if(Strings.isNullOrEmpty(childSql)){
                continue;
            }
            boolean result = false;
            for(IParser sqlParser : sqlParserList){
                if(!sqlParser.verify(childSql)){
                    continue;
                }

                sqlParser.parseSql(childSql, sqlTree);
                result = true;
            }

            if(!result){
                throw new RuntimeException("Syntax does not support.");
            }
        }

        if (preDealTable !=null) {
            for (CreateTableParser.SqlParserResult createTableParser : preDealTable) {
                sqlTree.addPreDealTableInfo(createTableParser.getTableName(), createTableParser);
            }
        }

        //解析exec-sql
        if(sqlTree.getExecSqlList().size() == 0){
            throw new RuntimeException("sql no executable statement");
        }

        for(InsertSqlParser.SqlParseResult result : sqlTree.getExecSqlList()){
            List<String> sourceTableList = result.getSourceTableList();
            List<String> targetTableList = result.getTargetTableList();

            Set<String> tmpTableList = sqlTree.getTmpTableMap().keySet();

            for(String tableName : sourceTableList){
                if (!tmpTableList.contains(tableName)){
                    CreateTableParser.SqlParserResult createTableResult = sqlTree.getPreDealTableMap().get(tableName);
                    if(createTableResult == null){
                        throw new RuntimeException("can't find table " + tableName);
                    }

                    TableInfo tableInfo = tableInfoParser.parseWithTableType(ETableType.SOURCE.getType(),
                            createTableResult);
                    sqlTree.addTableInfo(tableName, tableInfo);
                }
            }

            for(String tableName : targetTableList){
                if (!tmpTableList.contains(tableName)){
                    CreateTableParser.SqlParserResult createTableResult = sqlTree.getPreDealTableMap().get(tableName);
                    if(createTableResult == null){
                        throw new RuntimeException("can't find table " + tableName);
                    }

                    TableInfo tableInfo = tableInfoParser.parseWithTableType(ETableType.SINK.getType(),
                            createTableResult);
                    sqlTree.addTableInfo(tableName, tableInfo);
                }
            }
        }

        for (CreateTmpTableParser.SqlParserResult result : sqlTree.getTmpSqlList()){
            List<String> sourceTableList = result.getSourceTableList();
            for(String tableName : sourceTableList){
                if (!sqlTree.getTableInfoMap().keySet().contains(tableName)){
                    CreateTableParser.SqlParserResult createTableResult = sqlTree.getPreDealTableMap().get(tableName);
                    if(createTableResult == null){
                        if (sqlTree.getTmpTableMap().get(tableName) == null){
                            throw new RuntimeException("can't find table " + tableName);
                        }
                    }else {
                        TableInfo tableInfo = tableInfoParser.parseWithTableType(ETableType.SOURCE.getType(),
                                createTableResult);
                        sqlTree.addTableInfo(tableName, tableInfo);
                    }
                }
            }
        }

        return sqlTree;
    }

    public static List<String> getExecSqlTables(String sql){

        if(StringUtils.isBlank(sql)){
            throw new RuntimeException("sql is not null");
        }


        sql = sql.replaceAll("--.*", "")
                .replaceAll("\r\n", " ")
                .replaceAll("\n", " ")
                .replace("\t", " ").trim();

        List<String> sqlArr = DtStringUtil.splitIgnoreQuota(sql, SQL_DELIMITER);
        SqlTree sqlTree = new SqlTree();
        TableInfoParser tableInfoParser = new TableInfoParser();
        for(String childSql : sqlArr){
            if(Strings.isNullOrEmpty(childSql)){
                continue;
            }
            boolean result = false;
            for(IParser sqlParser : sqlParserList){
                if(!sqlParser.verify(childSql)){
                    continue;
                }

                sqlParser.parseSql(childSql, sqlTree);
                result = true;
            }

            if(!result){
                throw new RuntimeException("Syntax does not support.");
            }
        }
        //解析exec-sql
        if(sqlTree.getExecSqlList().size() == 0){
            throw new RuntimeException("sql no executable statement");
        }

        Set<String> tables = new LinkedHashSet();


        for(InsertSqlParser.SqlParseResult result : sqlTree.getExecSqlList()) {
            for (String table : result.getSourceTableList()) {
                tables.add(table);
            }
            for (String table : result.getTargetTableList()) {
                tables.add(table);
            }
        }

        for (CreateTmpTableParser.SqlParserResult result : sqlTree.getTmpSqlList()){
            for (String table : result.getSourceTableList()) {
                tables.add(table);
            }
        }

        for (String  tmpTable: sqlTree.getTmpTableMap().keySet()){
                tables.remove(tmpTable);
            }

        List<String> list = new ArrayList();
        for (String table : tables){
            list.add(table);
        }

        for (String table : sqlTree.getPreDealTableMap().keySet()){
            list.remove(table);
        }
        return list;
    }
}
